package pushserver

import (
	"errors"
	"fmt"
	"pushserver/connpool"
	"pushserver/msgqueue"
	"sync"
	"time"
)

var (
	Err_TaskIsNoDefine = errors.New("task is not define")
)

type SenderScheduler struct {
	Sender
	// 最大并行数
	MaxParallel int
	// 最小并行数
	MinParallel int
	// 队列堆积消息阀值，达到就增加并行
	QueueBufferLen int
	// 当前并行数
	parallenNum int
	task        func(*msgqueue.Message)
	sync.Mutex
}

// 初始化一个sender 调度器
func NewSender(p connpool.Pool, q msgqueue.Queue, proxy *TcpProxy) *SenderScheduler {
	this := &SenderScheduler{}
	this.pool = p
	this.queue = q
	this.TcpProxy = proxy
	//定义任务
	this.task = func(m *msgqueue.Message) {
		// 判断是否重发消息
		if m.TrySendTimes > 0 {
			sTime := m.SendTime.Add(time.Duration(this.ResendInterval) * time.Millisecond)
			if time.Now().Before(sTime) {
				this.queue.Push(m) // 未到重发时间，重新入队
				return
			}
		}
		if err := this.send(m); err != nil {
			if this.FailedCallback != nil { // 发送失败回调
				this.FailedCallback(err)
			}
			// 失败处理
			this.msgSendFailed(m)
		}
	}
	return this
}

func (this *SenderScheduler) BeginSender() error {
	if this.task == nil {
		return Err_TaskIsNoDefine
	}
	this.parallenNum = 0
	if this.MaxParallel <= 0 {
		this.MaxParallel = 10
	}
	if this.MinParallel <= 0 {
		this.MinParallel = 1
	}
	if this.QueueBufferLen <= 0 {
		this.QueueBufferLen = 1000
	}
	for i := 0; i < this.MinParallel; i++ {
		this.newGoroutine()
	}
	go func() {
		for {
			if this.queue.Count() >= this.QueueBufferLen {
				this.newGoroutine()
			}
			time.Sleep(time.Second * 10)
		}
	}()

	return nil
}

func (this *SenderScheduler) newGoroutine() {
	defer this.Unlock()
	this.Lock()
	// 判断是否达到最大并行
	if this.parallenNum >= this.MaxParallel {
		return
	} else {
		this.parallenNum++
	}
	go func(this *SenderScheduler) {
		defer func() {
			if err := recover(); err != nil {
				fmt.Println(err)
			}
		}()
		for {
			m, err := this.queue.Pop()
			if err == msgqueue.Err_QueueIsEmpty {
				this.Lock()
				b := this.parallenNum <= this.MinParallel
				this.Unlock()
				// 判断协程结束或阻塞
				if b {
					// 已达到最小并行数量 ，休眠
					time.Sleep(time.Second * 1)
				} else {
					// 如果消息队列为空而且 ，就退出循环结束routine
					break
				}
			} else if err == nil {
				this.task(m.(*msgqueue.Message))
			}
		}
	}(this)
}
